package com.lizhiyu.flink.demo3_sink.redis;

import com.lizhiyu.flink.model.VideoOrder;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import java.util.Date;
import java.util.Random;


/**
 * 使用apache插件 sink 向redis中存储信息  https://bahir.apache.org/docs/flink/current/flink-streaming-redis/
 * flink 官方没有对应的 redis 的 sink 的插件
 */
public class RedisApachebahirDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

        DataStream<VideoOrder> ds = env.addSource(new RichSourceFunction<VideoOrder>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("开始");
            }
            @Override
            public void run(SourceContext<VideoOrder> sourceContext) throws Exception {
                while(true){
                    Thread.sleep(1000l);
                    Random random = new Random();
                    int i = random.nextInt(10);
                    sourceContext.collect(new VideoOrder("1", "key:"+i, 1, 2019,new Date()));
                }
            }
            @Override
            public void cancel() {
                System.out.println("结束");
            }
        });

        //转换成Tuple
        DataStream<Tuple2<String, Integer>> mapDs = ds.map(new MapFunction<VideoOrder, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(VideoOrder videoOrder) throws Exception {
                return new Tuple2<String,Integer>(videoOrder.getTitle(),videoOrder.getMoney());
            }
        });
        //分组
        KeyedStream<Tuple2<String, Integer>, Object> keyDs = mapDs
                .keyBy(new KeySelector<Tuple2<String, Integer>, Object>() {
                    @Override
                    public Object getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0;
                    }
                });
        //统计
        //positionToSum   统计tuple的哪一列  下标从零开始
        SingleOutputStreamOperator sumDs = keyDs.sum(1);

        sumDs.print();

        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
                .setHost("39.107.109.94")
                .setPort(6379)
                .setPassword("Lizhiyu123")
                .build();

        //RedisSink<Tuple2<String, String>> tuple2RedisSink = new RedisSink(conf, new RedisExampleMapper());
        RedisSink<Tuple2<String, String>> tuple2RedisSink = new RedisSink(conf, new RedisExampleMapper());

        sumDs.addSink(tuple2RedisSink);
        env.execute();
    }
}
